package team.bluepen.supermarket.data.hbase;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import team.bluepen.supermarket.constant.Constant;
import team.bluepen.supermarket.data.entity.Good;
import team.bluepen.supermarket.data.hdfs.TableReadMapper;

/**
 * @author Kai
 */
public class GoodsLineReader implements LineReader<Good> {

    @Override
    public Good readLine(String line) {
        String[] splits = line.split("\t");
        if (splits.length <= 4) {
            return null;
        }
        try {
            return new Good()
                    .setId(Long.parseLong(splits[0]))
                    .setName(splits[1])
                    .setAmount(Integer.parseInt(splits[2]))
                    .setPrice(Integer.parseInt(splits[3]))
                    .setState(splits[4])
                    .generateRowKey();
        } catch (NumberFormatException e) {
            return null;
        }

    }

    @Override
    public Put buildPut(Good good) {
        Put put = new Put(Bytes.toBytes(good.getRow()));
        put.addColumn(Bytes.toBytes(Constant.GoodTable.TABLE_FAMILY_1), Bytes.toBytes(Constant.GoodTable.QUALIFIER_ID),
                Bytes.toBytes(good.getId()));
        put.addColumn(Bytes.toBytes(Constant.GoodTable.TABLE_FAMILY_1), Bytes.toBytes(Constant.GoodTable.QUALIFIER_NAME),
                Bytes.toBytes((good.getName() == null) ? "" : good.getName()));
        put.addColumn(Bytes.toBytes(Constant.GoodTable.TABLE_FAMILY_2), Bytes.toBytes(Constant.GoodTable.QUALIFIER_PRICE),
                Bytes.toBytes(good.getPrice()));
        put.addColumn(Bytes.toBytes(Constant.GoodTable.TABLE_FAMILY_2), Bytes.toBytes(Constant.GoodTable.QUALIFIER_AMOUNT),
                Bytes.toBytes(good.getAmount()));
        put.addColumn(Bytes.toBytes(Constant.GoodTable.TABLE_FAMILY_2), Bytes.toBytes(Constant.GoodTable.QUALIFIER_STATE),
                Bytes.toBytes(good.getState()));
        return put;
    }

    @Override
    public Put buildPut(String line) {
        return buildPut(readLine(line));
    }

    public static class GoodsTableMapper extends TableReadMapper<Good> {
        public GoodsTableMapper() {
            super(new GoodsLineReader());
        }
    }
}
